package com.adcc.ags.cgw.proxy;

import com.adcc.ags.cgw.configuration.AMQSFactory;
import com.adcc.ags.cgw.configuration.UserListFactory;
import com.adcc.ags.cgw.entity.mq.MQInfo;
import com.adcc.ags.cgw.entity.mq.MQState;
import com.adcc.ags.cgw.entity.msg.CGWMsg;
import com.adcc.ags.cgw.entity.msg.MsgType;
import com.adcc.utility.log.Log;
import com.adcc.utility.mq.configuration.MQConfiguration;
import com.adcc.utility.mq.configuration.MQConfigurationFactory;
import com.adcc.utility.mq.entity.Message;
import com.adcc.utility.mq.transfer.MQConnectionPool;
import com.adcc.utility.mq.transfer.MQConnectionPoolFactory;
import com.adcc.utility.mq.transfer.MQTransfer;
import com.adcc.utility.mq.transfer.ibm.IBMMQConnectionPool;
import com.adcc.utility.mq.transfer.ibm.IBMMQTransfer;
import com.google.common.collect.Lists;
import io.dropwizard.lifecycle.Managed;
import io.dropwizard.setup.Environment;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.format.DateTimeFormat;

import java.util.List;

/**
 * AMQS代理
 */
public class AMQSProxy implements Managed {

    // AMQS处理线程名称
    private static final String THREAD_NAME = "Thread.AMQSProcessor";

    // 单例对象
    private static AMQSProxy instance;

    // 接收队列
    private String recvQueue;

    // 发送队列
    private String sendQueue;

    // AMQS配置信息
    private List<ConnectionFactory> amqsFactoryList = Lists.newArrayList();

    // AMQS状态
    private MQInfo amqs = new MQInfo();

    // MQConfiguration
    private MQConfiguration configuration;

    // MQConnectionPool
    private MQConnectionPool connectionPool;

    // MQTransfer
    private MQTransfer transfer = new IBMMQTransfer();

    // AMQS处理线程
    private AMQSProcessor processor;

    /**
     * 构造方法
     * */
    private AMQSProxy() {

    }

    /**
     * 单例方法
     * */
    public static AMQSProxy getInstance() {
        if (instance == null) {
            instance = new AMQSProxy();
        }
        return instance;
    }

    public MQInfo getAMQS(){
        return amqs;
    }

    /**
     * 建立Managed
     * @param environment
     */
    public void build(Environment environment, AMQSFactory amqsFactory) throws Exception {
        if (environment == null) {
            throw new NullPointerException("environment is null");
        }
        if (amqsFactory == null) {
            throw new NullPointerException("downlinkFactory is null");
        }

        // 初始化MQCluster配置
        for(ConnectionFactory connection : amqsFactory.getAmqsFactoryList()){
            ConnectionFactory cf = new ConnectionFactory();
            cf.setHost(connection.getHost());
            cf.setPort(connection.getPort());
            cf.setHost(connection.getHost());
            cf.setQueueManager(connection.getQueueManager());
            cf.setChannel(connection.getChannel());
            cf.setRecvQueue(connection.getRecvQueue());
            cf.setSendQueue(connection.getSendQueue());
            amqsFactoryList.add(cf);
        }

        // 初始化MQ连接池
        if(amqsFactory.getMaxConnections() > 0 && amqsFactory.getMaxIdelConnections() > 0){
            connectionPool = MQConnectionPoolFactory.getInstance().createIBMMQConnectionPool();
            ((IBMMQConnectionPool)connectionPool).setActiveMode(amqsFactory.getActiveMode());
            ((IBMMQConnectionPool)connectionPool).setTimeout(amqsFactory.getTimeout());
            ((IBMMQConnectionPool)connectionPool).setMaxConnections(amqsFactory.getMaxConnections());
            ((IBMMQConnectionPool)connectionPool).setMaxIdelConnections(amqsFactory.getMaxIdelConnections());
        }
        environment.lifecycle().manage(this);
    }

    /**
     * 连接MQ集群
     */
    private void connectMQCluster() {
        try {

            // 循环连接AMQS MQ集群
            for (ConnectionFactory cf : amqsFactoryList) {
                recvQueue = cf.getRecvQueue();
                sendQueue = cf.getSendQueue();
                configuration = MQConfigurationFactory.getInstance().createIBMMQConfiguration(cf.getHost(), cf.getPort(), cf.getQueueManager(), cf.getChannel());
                transfer.setConfiguration(configuration);
                if(connectionPool != null){
                    connectionPool.init(configuration);
                }
                transfer.setConnectionPool(connectionPool);
                if (transfer.isConnected()) {
                    Log.info(AMQSProxy.class.getName(), "MQCluster connected host:" + cf.getHost() + " port:" + cf.getPort() + " channel:" + cf.getChannel() + " qm:" + cf.getQueueManager());
                    amqs.setHost(cf.getHost());
                    amqs.setPort(cf.getPort());
                    amqs.setChannel(cf.getChannel());
                    amqs.setQm(cf.getQueueManager());
                    amqs.setState(MQState.CONNECTED);
                    return;
                }
            }
            if(amqs.getState() != MQState.CONNECTING){
                amqs.setState(MQState.CONNECTING);
            }
        } catch (Exception ex) {
            Log.error(AMQSProxy.class.getName(), "connectMQCluster() error", ex);
            if(amqs.getState() != MQState.CONNECTING){
                amqs.setState(MQState.CONNECTING);
            }
        }
    }

    /**
     * 发送上行报文
     * @param msg
     * @throws Exception
     */
    public void sendUpLinkMsg(byte[] msg) throws Exception {
        try {
            Message message = new Message(msg);
            transfer.sendQueue(sendQueue, message);
        } catch (Exception ex) {
            Log.error(AMQSProxy.class.getName(), "sendUpLinkMsg() error", ex);
            throw ex;
        }
    }

    @Override
    public void start() throws Exception {
        try {

            // 连接MQCluster
            connectMQCluster();

            // 启动AMQS处理线程
            processor = new AMQSProcessor();
            new Thread(processor,THREAD_NAME).start();
        } catch (Exception ex) {
            Log.error(AMQSProxy.class.getName(), "start() error", ex);
        }
    }

    @Override
    public void stop() throws Exception {
        try {

            // 设置AMQS状态
            amqs.setState(MQState.DISCONNECTED);

            // 关闭下行处理线程
            if (processor != null) {
                processor.close();
            }
        } catch (Exception ex) {
            Log.error(AMQSProxy.class.getName(), "stop() error", ex);
        }
    }

    /**
     * 下行处理线程
     */
    class AMQSProcessor implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close() {
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(AMQSProcessor.class.getName(), "AMQSProcessor start...");
                isStarted = true;
                while (isStarted) {
                    try {
                        if (amqs.getState() == MQState.CONNECTED) {
                            if (UserListFactory.getInstance().getUserList().size() > 0) {
                                try{
                                    if(((IBMMQTransfer)transfer).getDepth(recvQueue) > 0){

                                        // 接收下行报文
                                        Message message = transfer.receiveQueue(recvQueue);
                                        if (message != null) {
                                            Log.info(AMQSProcessor.class.getName(), "receive downlink message\r\n" + message.toString());

                                            // 执行MsgProxy下行报文业务逻辑链
                                            CGWMsg cgwMsg = new CGWMsg(message, MsgType.DOWNLINK, "1");
                                            MsgProxy.getInstance().getMsgRouteActor().tell(cgwMsg,MsgProxy.getInstance().getMsgRouteActor());
                                        }
                                    }else {
                                        Thread.sleep(10);
                                    }
                                }catch (Exception ex){
                                    Log.error(AMQSProcessor.class.getName(), "receive downlink message error", ex);
                                    if(amqs.getState() != MQState.CONNECTING){
                                        amqs.setState(MQState.CONNECTING);
                                    }
                                }
                            }else{
                                Thread.sleep(100);
                            }
                        } else {

                            // 重连MQCluster
                            connectMQCluster();
                            Thread.sleep(100);
                        }
                    } catch (Exception ex) {
                        Log.error(AMQSProcessor.class.getName(), "process message error", ex);
                    }
                }
            } catch (Exception ex) {
                Log.error(AMQSProcessor.class.getName(), "run() error", ex);
            }
        }
    }
}
